package org.lql.chaos.kafka.example.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Output;
import org.apache.kafka.common.serialization.Serializer;
import org.lql.chaos.kafka.example.pojo.Order;
import org.lql.chaos.kafka.example.pojo.OrderItem;

import java.math.BigDecimal;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Map;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 03:26
 * @description:
 **/
public class KryoSerializer<T> implements Serializer<T> {

	// 使用ThreadLocal保证线程安全和每个线程独立Kryo实例
	private final ThreadLocal<Kryo> kryoThreadLocal = ThreadLocal.withInitial(() -> {
		Kryo kryo = new Kryo();
		// 注册常用类，确保序列化和反序列化时一致
		kryo.register(ArrayList.class);
		kryo.register(Order.class);
		kryo.register(OrderItem.class);
		kryo.register(BigDecimal.class);
		kryo.register(Timestamp.class);
		kryo.register(String.class);
		return kryo;
	});

	@Override
	public void configure(Map<String, ?> configs, boolean isKey) {
		// 可以空实现，注册已经在线程本地完成
	}

	@Override
	public byte[] serialize(String topic, T data) {
		if (data == null) return null;
		Kryo kryo = kryoThreadLocal.get();
		try (Output output = new Output(4096, -1)) {
			kryo.writeClassAndObject(output, data);
			return output.toBytes();
		}
	}

	@Override
	public void close() {
		// 资源清理可选
	}
}
